package com.gxljc.bear.crawler.weixin;

import com.mongodb.BasicDBObject;
import com.mongodb.Bytes;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.gxljc.commons.mongo.cli.BaseMongoRepository;
import com.gxljc.bear.crawler.base.BaseMongoTools;
import com.gxljc.bear.crawler.base.DolphinCrawlerConsts;
import com.gxljc.bear.crawler.util.SparkUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;

import java.io.Serializable;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 新榜微信PAGE数据爬虫。
 *
 * @author tanghaitao
 * @since 2022-10-10
 */
public class BizCrawler implements Serializable {
    public static Logger LOG = Logger.getLogger(BizCrawler.class);

    public BizCrawler() {
    }

    public void crawl() throws Exception {
        List<AccountTable> seeds = null;
        try {
            seeds = getSeed();
        } catch (Exception er) {
            er.printStackTrace();
        }
        int size = seeds.size();
        if (size < WeixinConst.MAX_SINGLE_CORE) { //少于一定数量时，使用单线程爬取
            try {
                crawlNoSpark(seeds);
            } catch (Exception er) {
                er.printStackTrace();
                LOG.error("single core job error", er);
            }
        } else {
            try {
                crawl(seeds);
            } catch (Exception er) {
                er.printStackTrace();
                LOG.error("spark job error", er);
            }
        }
    }

    public void crawl(List<AccountTable> seeds) throws Exception {
        int coreMax = 10;
        JavaSparkContext jsc = SparkUtil.createCommonsSparkContext(
                "bear-biz爬虫-" + seeds.size(), coreMax,
                coreMax * 2, BizCrawler.class);
        JavaRDD<AccountTable> seedsRDD = jsc.parallelize(new ArrayList<AccountTable>(seeds));
        long count = seedsRDD.mapPartitions(
                new FlatMapFunction<Iterator<AccountTable>, Integer>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Iterable<Integer> call(Iterator<AccountTable> seeds)
                            throws Exception {
                        int successCnt = 0;

                        while (seeds.hasNext()) {
                            AccountTable seed = seeds.next();
                            try {
                                int ret = crawl(seed);
                                if (ret == 1)
                                    successCnt++;
                                if (successCnt % 100 == 0) {
                                    LOG.info("successCnt=" + successCnt);
                                }
                            } catch (Exception er) {
                                er.printStackTrace();
                            }
                        }
                        return Arrays.asList(successCnt);
                    }
                }).count();
        LOG.info("all count=" + count);
        jsc.stop();
    }

    public void crawlNoSpark(List<AccountTable> seeds) {
        if (CollectionUtils.isEmpty(seeds)) {
            LOG.error("page is empty");
            return;
        }
        for (AccountTable table : seeds) {
            try {
                crawl(table);
            } catch (Exception e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            }
        }
    }

    //单条url入口
    public int crawl(AccountTable table) throws Exception {
        LOG.info("crawl = " + table.accountId);
        String url = getCrawlUrl(table.accountId);
        if(StringUtils.isEmpty(url)){
            LOG.info("url is empty no find biz " + table.accountId);
            return DolphinCrawlerConsts.RESULT_NO;
        }
        //http://mp.weixin.qq.com/s?__biz=MjUwNjU1NzA0MQ==&mid=2662157541&idx=1&sn=63e02a5747e1ef0ef5701937abc9e9ae&chksm=a6011dfb917694ed8058c124a7a179e9a373fb2e701ce32ed1d02002154f65c1f99b1b912820&scene=4
        Pattern p = Pattern.compile("_biz=(.*?)&");
        Matcher m = p.matcher(url);
        if (!m.find()) {
            LOG.info("no find biz " + table.accountId);
            return DolphinCrawlerConsts.RESULT_NO;
        }
        String biz = m.group(1);
//        LOG.info(biz);
        table.setBiz(biz);
//        LOG.info(table);
        WeixinMongodbUtil.saveMongodbAccount(table, WeixinConst.MONGODB_TABLE_ACCOUNT);
        return DolphinCrawlerConsts.RESULT_YES;
    }

    public String getCrawlUrl(String accountId) throws Exception {
        BaseMongoRepository repo = BaseMongoTools.getInstance(DolphinCrawlerConsts.MongoDBName.WEIXIN.getName());
        BasicDBObject keys = new BasicDBObject();
        keys.put(WeixinConst.SCHEMA_COLUMN_NAME.ORG_URL.getValue(), 1);
        DBObject query = new BasicDBObject();
        query.put(WeixinConst.SCHEMA_COLUMN_NAME.ACCOUNT_ID.getValue(), new BasicDBObject("$eq", accountId));
        DBCursor cursor = repo.getCollection(WeixinConst.MONGODB_TABLE_PAGE)
                .find(query, keys)
                .addOption(Bytes.QUERYOPTION_NOTIMEOUT);
        List<DBObject> dataIterator = cursor.toArray();
        for (DBObject data : dataIterator) {
            try {
                Object url = data.get(WeixinConst.SCHEMA_COLUMN_NAME.ORG_URL.getValue());
                if (url != null && !StringUtils.isEmpty((String) url)) {
                    if (((String) url).indexOf("biz") >= 0)
                        return (String) url;
                }
            } catch (Exception er) {
                er.printStackTrace();
            }
        }
        return null;
    }

    //获取seed
    public List<AccountTable> getSeed() throws Exception {
        BaseMongoRepository repo = BaseMongoTools.getInstance(DolphinCrawlerConsts.MongoDBName.WEIXIN.getName());
        BasicDBObject keys = new BasicDBObject();
        keys.put(DolphinCrawlerConsts.MONGODB_DEFAULT_ID, 1);
        keys.put(WeixinConst.SCHEMA_COLUMN_NAME.ACCOUNT_ID.getValue(), 1);
        DBObject query = new BasicDBObject();
        query.put(WeixinConst.SCHEMA_COLUMN_NAME.BIZ.getValue(), new BasicDBObject("$eq", null));
        DBCursor cursor = repo.getCollection(WeixinConst.MONGODB_TABLE_ACCOUNT)
                .find(query, keys)
                .addOption(Bytes.QUERYOPTION_NOTIMEOUT);
        List<AccountTable> seeds = new LinkedList<>();
        List<DBObject> dataIterator = cursor.toArray();
        for (DBObject data : dataIterator) {
            try {
                Object id = data.get(DolphinCrawlerConsts.MONGODB_DEFAULT_ID);
                Object accountId = data.get(WeixinConst.SCHEMA_COLUMN_NAME.ACCOUNT_ID.getValue());
                AccountTable table = new AccountTable();
                table.setAccountId(accountId.toString());
                System.out.println(accountId.toString());
                seeds.add(table);
            } catch (Exception er) {
                er.printStackTrace();
            }
        }
        return seeds;
    }
}
